-- drop function if exists sm_rdd.ft_rdd_map_output(int, int);
create or replace function sm_rdd.ft_rdd_map_output
(
  i_job_no         int           ,
  i_timeout        int                -- 超时时间，单位：毫秒
)
returns table
(
  o_key            varchar(64)   ,
  o_frame          text
)
as
$$
declare --here
  v_timer               timestamp := now();


begin

  
  -- 限时等待各个 work 节点 任务轮询器执行，返回结果

  -- -- -- sql 和 while 语句还没能实现 Asynchronous notification 的监听阻塞
  -- -- -- 此处 while 轮询，可以用 c 或 rust 改造为由 listen-notify 的 listen 和 dblink_get_notify 触发，示例如下：
  -- -- -- -- select dblink_connect('conn', 'hostaddr=192.168.3.105 port=5432 dbname=postgres user=postgres password=seanguo_007')
  -- -- -- -- select dblink_exec('conn', 'listen aaa');
  -- -- -- -- select * from dblink_get_notify('conn');
  -- -- -- -- notify aaa, 'im bbb'
  -- -- -- -- notify aaa, 'im ccc'
  -- -- -- -- notify aaa, 'im ddd'
  -- -- -- -- select * from dblink_get_notify('conn');   -- 阅后即焚
  -- -- -- -- select dblink_exec('conn', 'unlisten aaa');
  -- -- -- -- select dblink_disconnect('conn')

  while not exists (select from sm_rdd.tb_trs_distrb_job where job_no = i_job_no and job_done_work_cnt = job_map_work_cnt)
    and not exists (select from sm_rdd.tb_parted_trs_work_list where work_job_no = i_job_no and work_status = 'failed')
  loop
    if now() >= v_timer + (interval '1 millisecond' * (i_timeout + 50))
    then
      raise exception 'error: time out beyond % millisecond', (i_timeout + 50)::text;
    end if;
    perform pg_sleep(0.05);
  end loop
  ;

  if not exists (select from sm_rdd.tb_parted_trs_work_list where work_job_no = i_job_no and work_status = 'failed')
  then
    return query
      -- -- select 
      -- --   dat.reduce_key           as o_key             ,
      -- --   dat.work_output_frame    as o_frame
      -- -- from sm_rdd.tb_parted_trs_work_list lst
      -- -- inner join sm_rdd.tb_sour_work_data dat
      -- --   on dat.work_no = lst.work_no
      -- -- where lst.work_job_no = i_job_no
    
      select 
        dat.reduce_key           as o_key             ,
        dat.work_output_frame    as o_frame
      from sm_rdd.tb_sour_work_data dat
      where dat.work_job_no_ex = i_job_no
      order by dat.reduce_key
    ;
  else
    raise exception 'failed job_no: %! ', i_job_no::text;
  end if;
  
  exception
  when others then    
  raise exception '(%)', sqlerrm; 
  
end
$$
language plpgsql volatile
cost 100;

-- select o_key,o_frame 
-- from sm_rdd.ft_rdd_map_output
-- (
--   i_job_no        => ...
--   , i_timeout     => 500
-- )